package cn.doitedu.rtdw.features_etl;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/30
 * @Desc: 学大数据，上多易教育
 *
 *
 **/
public class 广告转化率预估_特征数据加工 {
    public static void main(String[] args) {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 创建逻辑表（连接器表），映射kafka中物理topic（维度打宽明细行为日志）
         */
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     url_prefix    STRING,                         "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     proc_time AS proctime(),                      " // processing time 时间语义
                        + "     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        + "     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd_events',                          "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        // 提取需求所要的事实数据
        tenv.executeSql(
                " create temporary view ad_events  AS                   "+
                        " select                                                 "+
                        "   user_id,                                             "+
                        " 	event_time,                                          "+
                        " 	event_id,                                            "+
                        " 	properties['ad_id'] as ad_id,                        "+
                        " 	properties['ad_tracking_id'] as ad_tracking_id,      "+
                        " 	rt                                                   "+
                        " from dwd_kafka                                         "+
                        " where event_id in ('ad_show','ad_click','ad_transfer') "
        );

        // 用cep做模式（规则）匹配，  有曝光、后面有1次或多次点击，后面有转化
        // 匹配到后，输出：  用户id,ad_id,曝光时间,点击时间,转化时间
        tenv.executeSql(
                "select                                               "+
                        "    uid,adid,tkid,show_time,click_time,transfer_time  "+
                        "from ad_events                                        "+
                        "    MATCH_RECOGNIZE(                                  "+
                        "	    PARTITION BY ad_tracking_id                   "+
                        "	    ORDER BY rt                                   "+
                        "	    MEASURES                                      "+
                        "	        A.user_id as uid,                         "+
                        "	        A.ad_id as adid,                          "+
                        "	        A.ad_tracking_id as tkid,                 "+
                        "	        A.event_time as show_time,                "+
                        "	        FIRST(B.event_time) as click_time,        "+
                        "	        C.event_time as transfer_time             "+
                        "	    ONE ROW PER MATCH                             "+
                        "	    AFTER MATCH SKIP PAST LAST ROW                "+
                        "	    PATTERN (A B+ C) WITHIN INTERVAL '1' MINUTE   "+
                        "	    DEFINE                                        "+
                        "	        A AS A.event_id = 'ad_show',              "+
                        "	    	B AS B.event_id = 'ad_click',             "+
                        "	    	C AS C.event_id = 'ad_transfer'           "+
                        "                                                     "+
                        "	)                                                 "
        ).print();
    }
}
